package com.guoj.rabbitmq.sender;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.guoj.rabbitmq.entity.BrokerMessageLog;
import com.guoj.rabbitmq.entity.TOrder;
import com.guoj.rabbitmq.service.IBrokerMessageLogService;
import com.guoj.rabbitmq.service.impl.BrokerMessageLogServiceImpl;
import com.guoj.rabbitmq.utils.Constant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Date;

/**
 * @作者: guoj
 * @日期: 2019/5/30 15:51
 * @描述: 消息可靠投递-生产方
 */
@Component
@Slf4j
public class RabbitOrderSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

	@Autowired
	private RabbitTemplate rabbitTemplate;
//	@Autowired
//	private IBrokerMessageLogService brokerMessageLogService;
	@Autowired
	private BrokerMessageLogServiceImpl brokerMessageLogService2;

	/**
	 * 发送消息, 构建自定义对象消息
	 * @param order
	 */
	public void sendOrder(TOrder order) {
		rabbitTemplate.setConfirmCallback(this);
		rabbitTemplate.setReturnCallback(this);

		/**
		 * 自定义对象，消息唯一ID，
		 * 每个发送的消息都需要配备一个 CorrelationData 相关数据对象，
		 * CorrelationData 对象内部只有一个 id 属性，用来表示当前消息唯一性。
		 * 真实场景是需要做业务无关消息 ID 生成，同时要记录下这个 id 用来纠错和对账。
 		 */
		CorrelationData correlationData = new CorrelationData(order.getMessageId());
		// 发送消息
		rabbitTemplate.convertAndSend(Constant.ORDER_EXCHANGE_ERROR, Constant.ORDER_ROUTING, order, correlationData);
		log.info("消息已发送，messageId={}", order.getMessageId());
	}

	/**
	 * 成功接收后回调, 确认消息被rabbitmq成功接收
	 */
	@Override
	public void confirm(CorrelationData correlationData, boolean ack, String s) {
		String messageId = correlationData.getId();
		if (ack) {
			// 如果成功接收了，就修改消息记录表的状态为success。
			BrokerMessageLog brokerMessageLog = new BrokerMessageLog();
			brokerMessageLog.setStatus(Constant.ORDER_SEND_SUCCESS);
			brokerMessageLog.setUpdateTime(LocalDateTime.now());
			QueryWrapper<BrokerMessageLog> queryWrapper = new QueryWrapper<BrokerMessageLog>();
			queryWrapper.eq("message_id", messageId);
			brokerMessageLogService2.update(brokerMessageLog, queryWrapper);
			log.info("消息发送成功, messageId={}", messageId);
		} else {
			//失败则进行具体的后续操作:重试 或者补偿等手段
			log.error("消息发送失败, messageId={}", messageId);
		}
	}

	/**
	 * 失败后回调
	 */
	@Override
	public void returnedMessage(Message message, int i, String s, String s1, String s2) {
		byte[] body = message.getBody();
		MessageProperties messageProperties = message.getMessageProperties();
		log.error("消息发送失败, body={}", new String(body));
	}
}
